package com.shujia.batch.car

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object DwsAppVioWsjtj {
  def main(args: Array[String]): Unit = {

    /**
      * 需求:
      * 若某设备一段时间均无违法数据上报，则认为该设备可能异常。通过非现场违法数据统计历史数据判断。
      * 过车数据中对设备进行分组,然后进行排名,
      * 然后对排名进行关联获取当前时间和下一时间之间的时间差 大于多少天被判定为设备异常
      *
      * 001,2018-05-01
      * 001,2018-05-03
      * 001,2018-05-08
      * 001,2018-09-01
      *
      * 1、取出有数据的时间
      * 2、按照时间排序，计算两个时间差
      * 3、盗用箱线图计算得出设备异常时间段
      *
      * 口径：
      * 一段时间是多长时间  用箱线图
      *
      */

    val spark: SparkSession = SparkSession
      .builder()
      .appName("DwsAppAcdMrjq")
      .enableHiveSupport() //开启hive元数据支持
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._


    //非现场违法表
    val vioSurveil: DataFrame = spark
      .table("dwd.dwd_base_vio_surveil")
      .where($"ds" === "20220627")

    //取出有数据设备
    val resultDF: DataFrame = vioSurveil
      .select(
        //违法日期
        date_format($"wfsj", "yyyy-MM-dd") as "wfsj",
        //设备编号
        $"sbbh"
      )
      //过滤设备编码为空的数据
      .where($"sbbh".isNotNull and $"sbbh" =!= "" and trim($"sbbh") =!= "-")
      //去除重复数据
      .distinct()
      //取出上一次有数据日期
      .withColumn("last_date", lag($"wfsj", 1) over Window.partitionBy($"sbbh").orderBy($"wfsj"))
      //计算当前日期和上一次有数据日期的时间差
      .withColumn("diff_sj", datediff($"wfsj", $"last_date"))
      //过滤掉时间差为空的数据
      .where($"diff_sj".isNotNull)
      //计算总的有数据的天数
      .withColumn("sum__date", count($"sbbh") over Window.partitionBy($"sbbh"))
      //计算Q1和Q3的下标
      .withColumn("Q1_index", floor($"sum__date" / 4))
      .withColumn("Q3_index", ceil($"sum__date" / 4 * 3))
      //在同一个设备，按照间隔天数排名
      .withColumn("wf_rank", row_number() over Window.partitionBy($"sbbh").orderBy($"diff_sj"))
      //取出Q1和Q3位置对应的值
      .withColumn("Q1", sum(when($"Q1_index" === $"wf_rank", $"diff_sj").otherwise(0)) over Window.partitionBy($"sbbh"))
      .withColumn("Q3", sum(when($"Q3_index" === $"wf_rank", $"diff_sj").otherwise(0)) over Window.partitionBy($"sbbh"))
      //计算IQR
      .withColumn("IQR", $"Q3" - $"Q1")
      //计算上边界
      .withColumn("max_value", $"Q3" + $"IQR" * 1.5)
      //取出异常设备时间段
      .where($"diff_sj" > $"max_value")
      //整理数据
      .select(
      $"sbbh",
      $"diff_sj" as "sjc",
      $"wfsj",
      $"last_date",
      round($"diff_sj" / $"max_value", 2) as "more_zb"
    )

    //保存数据
    resultDF.createOrReplaceTempView("tmp")
    spark.sql(
      """
        |insert overwrite table dws.dws_app_vio_wsjtj partition (ds='20220627')
        |select * from tmp
      """.stripMargin)


  }

}
